package com.atguigu.wuliu.utils;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.wuliu.app.func.DimAsyncByForeignKeyFunction;
import com.atguigu.wuliu.bean.DwsTransOrgReceiveDayBean;
import com.atguigu.wuliu.common.WuliuConfig;
import io.debezium.ddl.parser.mysql.generated.MySqlParser;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.CompletableFuture;

/**
 * HBase工具类
 */
public class HbaseUtil {

    //获取连接
    public static Connection getHbaseConnection() {
        Configuration conf = new Configuration();
        // 配置 Zookeeper
        conf.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104");

        try {
            Connection connection = ConnectionFactory.createConnection(conf);
            return connection;
        } catch (IOException e) {
            throw new RuntimeException();
        }
    }

    //关闭连接
    public static void closeHbaseConnection(Connection conn) {
        if (conn != null && !conn.isClosed()) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    //建表
    public static void createTable(Connection conn, String namespace, String tableName, String... families) {
        if (families.length < 1) {
            System.out.println("建表至少需要一个列族");
            return;
        }

        try (Admin admin = conn.getAdmin()) {
            TableName tableNameObj = TableName.valueOf(namespace, tableName);
            if (admin.tableExists(tableNameObj)) {
                System.out.println("要创建" + namespace + ":" + tableName + "的表已经存在");
                return;
            }
            System.out.println("在hbase中创建" + namespace + ":" + tableName + "表");
            TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableNameObj);
            for (int i = 0; i < families.length; i++) {
                String family = families[i];
                ColumnFamilyDescriptor columnFamilyDescriptor
                        = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(family)).build();
                tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptor);
            }
            admin.createTable(tableDescriptorBuilder.build());
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    //删表
    public static void dropTable(Connection conn, String namespace, String tableName) {
        try (Admin admin = conn.getAdmin()) {
            TableName tableNameObj = TableName.valueOf(namespace, tableName);
            if (!admin.tableExists(tableNameObj)) {
                System.out.println("要删除的" + namespace + ":" + tableName + "表不存在");
                return;
            }
            System.out.println("删除" + namespace + ":" + tableName + "表");
            admin.disableTable(tableNameObj);
            admin.deleteTable(tableNameObj);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //从hbase表中删除一行数据
    public static void delRow(Connection conn, String namespace, String tableName, String rowkey) throws IOException {
        TableName tableNameObj = TableName.valueOf(namespace, tableName);
        Table table = conn.getTable(tableNameObj);
        Delete delete = new Delete(Bytes.toBytes(rowkey));
        table.delete(delete);
        table.close();
    }

    //向hbase表中put一行数据
    public static void putRow(Connection conn, String namespace, String tableName, String rowkey, String family, String[] columns, String[] values) throws IOException {
        TableName tableNameObj = TableName.valueOf(namespace, tableName);
        Table table = conn.getTable(tableNameObj);
        Put put = new Put(Bytes.toBytes(rowkey));
        for (int i = 0; i < columns.length; i++) {
            String column = columns[i];
            String value = values[i];
            if (value != null) {
                put.addColumn(Bytes.toBytes(family), Bytes.toBytes(column), Bytes.toBytes(value));
            }
        }
        table.put(put);
        table.close();
    }


    //异步从hbase中查询维度数据
    public static JSONObject getAsyncDimInfoFromHbase(AsyncConnection asyncConnection, String nameSpace, String tableName, String rowKey) {
        TableName tableNameObj = TableName.valueOf(nameSpace, tableName);
        AsyncTable<AdvancedScanResultConsumer> asyncTable = asyncConnection.getTable(tableNameObj);
        Get get = new Get(Bytes.toBytes(rowKey));
        try {
            Result result = asyncTable.get(get).get();
            List<Cell> cells = result.listCells();
            if(cells != null && cells.size() > 0){
                JSONObject jsonObj = new JSONObject();
                for (Cell cell : cells) {
                    String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String columnValue = Bytes.toString(CellUtil.cloneValue(cell));
                    jsonObj.put(columnName,columnValue);
                }
                return jsonObj;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }
    // 根据外键从hbase表中异步查询一行数据
    public static JSONObject getRowByForeignKey(AsyncConnection asyncConnection,String namespace, String tableName, Tuple2<String,String> foreignKeyNameAndKey){
        AsyncTable<AdvancedScanResultConsumer> asyncTable = null;
        JSONObject dimJsonObj = null;
        try {
            asyncTable = asyncConnection.getTable(TableName.valueOf(namespace, tableName));
            Scan scan = new Scan();
            String foreignKeyName = foreignKeyNameAndKey.f0;
            String foreignKeyValue = foreignKeyNameAndKey.f1;
            SingleColumnValueFilter singleColumnValueFilter
                    = new SingleColumnValueFilter(Bytes.toBytes("info"),Bytes.toBytes(foreignKeyName), CompareOperator.EQUAL,Bytes.toBytes(foreignKeyValue));

            singleColumnValueFilter.setFilterIfMissing(true);
            scan.setFilter(singleColumnValueFilter);

            ResultScanner scanner = asyncTable.getScanner(scan);
            Result result = scanner.next();
            if(result != null){
                Cell[] cells = result.rawCells();
                if(cells.length > 0){
                    dimJsonObj = new JSONObject();
                    dimJsonObj.put("id",Bytes.toString(result.getRow()));
                    for (Cell cell : cells) {
                        String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
                        String columnValue = Bytes.toString(CellUtil.cloneValue(cell));
                        dimJsonObj.put(columnName,columnValue);
                    }
                }
            }else{
                System.out.println("从hbase表中没有找到对应的维度数据");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return dimJsonObj;
    }


    //创建异步hbase连接
    public static AsyncConnection getAsyncConnection() {
        Configuration conf = new Configuration();
        // 配置 Zookeeper
        conf.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104");
        try {
            return ConnectionFactory.createAsyncConnection(conf).get();
        } catch (Exception e) {
            throw new RuntimeException();
        }

    }

    //关闭异步连接
    public static void closeAsyncConnection(AsyncConnection conn) {
        if (conn != null && !conn.isClosed()) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }


    public static void main(String[] args) {
        AsyncConnection asyncConnection = getAsyncConnection();
        JSONObject rowByForeignKey = getRowByForeignKey(asyncConnection, WuliuConfig.HBASE_NAMESPACE, "dim_truck_info", Tuple2.of("id", "962"));
        System.out.println(rowByForeignKey);
        closeAsyncConnection(asyncConnection);
    }


}
